package com.guqj.mq.kafka;

import java.util.HashMap;
import java.util.Map;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerRecord;

public class Produder {

	private Producer<String, Object> producer;

	public Produder() {
		Map<String, Object> props = new HashMap<>();
		props.put("metadata.broker.list", KafkaConstant.KAFKA_ADDRESS);
		// key.serializer.class默认为serializer.class
		props.put("key.serializer", KafkaConstant.KAFKA_SERIALIZER_CLASS);
		props.put("value.serializer", KafkaConstant.KAFKA_SERIALIZER_CLASS);
		// 可选配置，如果不配置，则使用默认的partitioner
		props.put("partitioner.class", "com.guqj.kafka.PartitionerDemo");
		// 触发acknowledgement机制，否则是fire and forget，可能会引起数据丢失
		// 值为0,1,-1,可以参考
		props.put("bootstrap.servers", KafkaConstant.KAFKA_ADDRESS);
		props.put("request.required.acks", "1");

		producer = new KafkaProducer<String, Object>(props);
	}

	public void produce() {
		ProducerRecord<String, Object> record = null;
		for (int i = 0; i <= 255; i++) {
			String ip = "192.168.2." + i;
			String message = System.currentTimeMillis() + "|www.example.com|" + ip;
			record = new ProducerRecord<String, Object>(KafkaConstant.KAFKA_TOPIC, 0, ip, message);
			producer.send(record);
			System.out.println(record);
		}
		producer.close();
	}

	public static void main(String[] args) {
		new Produder().produce();
	}
}
